package com.fsql.config;

import com.fsql.utils.FlinkRestClientUtils;
import com.mysql.jdbc.StringUtils;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FlinkRestClientConfiguration {

    @Value("${flink.jobmanager.port}")
    private Integer jobManagerPort;

    @Value("${flink.jobmanager.host}")
    private String jobManagerHost;

    @Value("${flink.rest.port}")
    private Integer restPort;

    @Value("${flink.clusterId}")
    private String clusterId;

    @Bean
    public FlinkRestClientUtils initFlinkRestClient() {
        try {
            org.apache.flink.configuration.Configuration configuration = new org.apache.flink.configuration.Configuration();
            configuration.setString(JobManagerOptions.ADDRESS, jobManagerHost);

            if (jobManagerPort == null) {
                jobManagerPort = 6123;
            }
            configuration.setInteger(JobManagerOptions.PORT, jobManagerPort);
            configuration.setInteger(RestOptions.PORT, restPort);

            // rest客户端
            if (StringUtils.isNullOrEmpty(clusterId)) {
                clusterId = "flink-default";
            }
            RestClusterClient restClient = new RestClusterClient(configuration, clusterId);

            return new FlinkRestClientUtils(restClient);

        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;

    }
}
